package com.larry.spark.test

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}


/**
 * 编号	字段名称	字段类型	字段含义
 * 1	date	String	用户点击行为的日期
 * 2	user_id	Long	用户的ID
 * 3	session_id	String	Session的ID
 * 4	page_id	Long	某个页面的ID
 * 5	action_time	String	动作的时间点
 * 6	search_keyword	String	用户搜索的关键词
 * 7	click_category_id	Long	某一个商品品类的ID
 * 8	click_product_id	Long	某一个商品的ID
 * 9	order_category_ids	String	一次订单中所有品类的ID集合
 * 10	order_product_ids	String	一次订单中所有商品的ID集合
 * 11	pay_category_ids	String	一次支付中所有品类的ID集合
 * 12	pay_product_ids	String	一次支付中所有商品的ID集合
 * 13	city_id	Long	城市 id
 *
 */
object Top10_2 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
    val sc = new SparkContext(conf)

    val dataLine = sc.textFile("data/user_visit_action.txt")
    //缓存
//    dataLine.cache()

    //2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_37_2019-07-17 00:00:02_手机_-1_-1_null_null_null_null_3
    //点击，下单，支付
    //点击数据


    //点击 => (id,(1,0,0))
    //下单 => (id,(0,1,0))
    //支付 => (id,(0,0,1))
    val datas: RDD[(String, (Int, Int, Int))] = dataLine.flatMap(
      line => {
        val l = line.split("_")
        if (l(6) != "-1") {
          List((l(6), (1, 0, 0)))
        } else if (l(8) != "null") {
          val ids = l(8).split(",")
          ids.map(
            id => {
              (id, (0, 1, 0))
            }
          )
        } else if (l(10) != "null") {
          val ids = l(10).split(",")
          ids.map(
            id => {
              (id, (0, 0, 1))
            }
          )
        } else {
          Nil
        }
      }
    )

    val reduceData = datas.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )

    //排序 取前十
    val top = reduceData.sortBy(_._2, false).take(10)

    top.foreach(println)

    sc.stop()
  }

}
